package cn.cutie.clot.mq.server;

import cn.cutie.clot.mq.model.ClotMessage;
import cn.cutie.clot.mq.store.Indexer;
import cn.cutie.clot.mq.store.MessageStore;
import org.apache.catalina.Store;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @Description:
 * @Author: Cutie
 * @CreateDate: 2024/7/10 10:21
 * @Version: 0.0.1
 */
public class MessageQueue {
    public static final Map<String, MessageQueue> queues = new HashMap();

    private static final String TEST_TOPIC = "cn.cutie.test";

    static {
        queues.put(TEST_TOPIC, new MessageQueue(TEST_TOPIC));
        queues.put("a", new MessageQueue("a"));
    }

    // 订阅关系
    private Map<String, MessageSubscription> subscriptions = new HashMap<>();

    private String topic;
    // 满了就再加一个queue（或者文件）
//    private ClotMessage<?>[] queue = new ClotMessage[1024 * 10];
    private MessageStore store = null;
    private int index = 0;

    public MessageQueue(String topic) {
        this.topic = topic;
        store = new MessageStore(topic);
        store.init();
    }

    public static List<ClotMessage<?>> batch(String topic, String consumerId, int size) {
        MessageQueue messageQueue = queues.get(topic);
        if (messageQueue == null) throw new RuntimeException("topic not found");
        if (messageQueue.subscriptions.containsKey(consumerId)){

            int offset = messageQueue.subscriptions.get(consumerId).getOffset();

            int nextOffset = 0;
            // 如果不是第一条消息，需要更新offset
            if (offset > -1) {
                // 这里要取下一个消息的offset
                Indexer.Entry entry = Indexer.getEntry(topic, offset);
                nextOffset = offset + entry.getLength();
            }
            ClotMessage<?> recv = messageQueue.recv(nextOffset);

            List<ClotMessage<?>> result = new ArrayList<>();
            while(recv != null){
                result.add(recv);
                if (result.size() >= size) break;
                recv = messageQueue.recv(++offset); // TODO: 2024/7/15 这里不能++
            }
            System.out.println("===> batch topic/cid/size = " + topic + "/" + consumerId + "/" + result.size());
            System.out.println("===> batch last message = " + recv);
//            System.out.println("===> batch message = " + recv);
            return result;
        }
        throw new RuntimeException("subscriptions not found for topic/consumer = "
                + topic + "/" + consumerId);
    }

    public int send(ClotMessage<?> message) {
//        if (index >= queue.length) {
//            return -1;
//        }
//        message.getHeaders().put("X-offset", String.valueOf(index));
//        queue[index++] = message;

        int offset = store.pos();
        message.getHeaders().put("X-offset", String.valueOf(offset));
        store.write(message);
        return offset;
    }

    public ClotMessage<?> recv(int offset) {
//        if (ind <= index){
//            return queue[ind];
//        }
//        return null;

        return store.read(offset);
    }

    public void subscribe(MessageSubscription subscription){
        String consumerId = subscription.getConsumerId();
        subscriptions.putIfAbsent(consumerId, subscription);
    }

    public static void sub(MessageSubscription subscription){
        MessageQueue messageQueue = queues.get(subscription.getTopic());
        System.out.println("===> sub subscription = " + subscription);
        if (messageQueue == null) throw new RuntimeException("topic not found");
        messageQueue.subscribe(subscription);
    }

    public static int send(String topic, ClotMessage<String> message){
        MessageQueue messageQueue = queues.get(topic);
        if (messageQueue == null) throw new RuntimeException("topic not found");
        System.out.println("===> send topic/message = " + topic + "/" + message);
        return messageQueue.send(message);
    }

    public static ClotMessage<?> recv(String topic, String consumerId, int offset){
        MessageQueue messageQueue = queues.get(topic);
        if (messageQueue == null) throw new RuntimeException("topic not found");
        if (messageQueue.subscriptions.containsKey(consumerId)){
            return messageQueue.recv(offset);
        }
        throw new RuntimeException("subscriptions not found for topic/consumer = "
                + topic + "/" + consumerId);
    }

    // 使用此方法，需要手工调用ack，更新offset
    public static ClotMessage<?> recv(String topic, String consumerId){
        MessageQueue messageQueue = queues.get(topic);
        if (messageQueue == null) throw new RuntimeException("topic not found");
        if (messageQueue.subscriptions.containsKey(consumerId)){
            int offset = messageQueue.subscriptions.get(consumerId).getOffset();
            int nextOffset = 0;
            // 如果不是第一条消息，需要更新offset
            if (offset > -1) {
                // 这里要取下一个消息的offset
                Indexer.Entry entry = Indexer.getEntry(topic, offset);
                nextOffset = offset + entry.getLength();
            }
            ClotMessage<?> recv = messageQueue.recv(nextOffset);
            System.out.println("===> recv topic/cid/nextOffset = " + topic + "/" + consumerId + "/" + nextOffset);
            System.out.println("===> recv message = " + recv);
            return recv;
        }
        throw new RuntimeException("subscriptions not found for topic/consumer = "
                + topic + "/" + consumerId);
    }

    public static int ack(String topic, String consumerId, int offset){
        MessageQueue messageQueue = queues.get(topic);
        if (messageQueue == null) throw new RuntimeException("topic not found");
        if (messageQueue.subscriptions.containsKey(consumerId)){
            MessageSubscription messageSubscription = messageQueue.subscriptions.get(consumerId);
            if (offset > messageSubscription.getOffset() && offset < MessageStore.LENGTH){
                System.out.println("===> ack topic/cid/offset = " + topic + "/" + consumerId + "/" + offset);
                messageSubscription.setOffset(offset);
                return offset;
            }
            return -1;
        }
        throw new RuntimeException("subscriptions not found for topic/consumer = "
                + topic + "/" + consumerId);
    }

    public static void unSub(MessageSubscription subscription){
        System.out.println("===> unSub subscription = " + subscription);
        MessageQueue messageQueue = queues.get(subscription.getTopic());
//        if (messageQueue == null) return;
        if (messageQueue == null) throw new RuntimeException("topic not found");
        messageQueue.unsubscribe(subscription);
    }

    private void unsubscribe(MessageSubscription subscription) {
        String consumerId = subscription.getConsumerId();
        subscriptions.remove(consumerId);
    }
}
